package org.spiderflow.core.executor.shape;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
import com.google.common.hash.Funnels;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spiderflow.common.enums.PauseEvent;
import org.spiderflow.common.executor.Grammerable;
import org.spiderflow.common.context.CookieContext;
import org.spiderflow.common.context.SpiderContext;
import org.spiderflow.core.executor.function.MD5FunctionExecutor;
import org.spiderflow.core.http.HttpRequest;
import org.spiderflow.core.http.HttpResponse;
import org.spiderflow.core.utils.ExpressionUtils;
import org.spiderflow.common.executor.ShapeExecutor;
import org.spiderflow.common.respose.SpiderResponse;
import org.spiderflow.common.listener.SpiderListener;
import org.spiderflow.common.model.Grammar;
import org.spiderflow.common.model.SpiderNode;
import org.spiderflow.common.utils.DgbObjects;
import org.spiderflow.common.utils.StringLock;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * 请求执行器
 *
 * @author Administrator
 */
@Component
public class RequestExecutor implements ShapeExecutor, Grammerable, SpiderListener {

    private static final String SLEEP = "sleep";

    private static final String URL = "url";

    private static final String PROXY = "proxy";

    private static final String REQUEST_METHOD = "method";

    private static final String PARAMETER_NAME = "parameter-name";

    private static final String PARAMETER_VALUE = "parameter-value";

    private static final String COOKIE_NAME = "cookie-name";

    private static final String COOKIE_VALUE = "cookie-value";

    private static final String PARAMETER_FORM_NAME = "parameter-form-name";

    private static final String PARAMETER_FORM_VALUE = "parameter-form-value";

    private static final String PARAMETER_FORM_FILENAME = "parameter-form-filename";

    private static final String PARAMETER_FORM_TYPE = "parameter-form-type";

    private static final String BODY_TYPE = "body-type";

    private static final String BODY_CONTENT_TYPE = "body-content-type";

    private static final String REQUEST_BODY = "request-body";

    private static final String HEADER_NAME = "header-name";

    private static final String HEADER_VALUE = "header-value";

    private static final String TIMEOUT = "timeout";

    private static final String RETRY_COUNT = "retryCount";

    private static final String RETRY_INTERVAL = "retryInterval";

    private static final String RESPONSE_CHARSET = "response-charset";

    private static final String FOLLOW_REDIRECT = "follow-redirect";

    private static final String TLS_VALIDATE = "tls-validate";

    private static final String LAST_EXECUTE_TIME = "__last_execute_time_";

    private static final String COOKIE_AUTO_SET = "cookie-auto-set";

    private static final String REPEAT_ENABLE = "repeat-enable";

    private static final String BLOOM_FILTER_KEY = "_bloom_filter";

    @Value("${spider.workspace}")
    private String workspace;

    @Value("${spider.bloomfilter.capacity:5000000}")
    private Integer capacity;

    @Value("${spider.bloomfilter.error-rate:0.00001}")
    private Double errorRate;

    private static final Logger LOGGER = LoggerFactory.getLogger(RequestExecutor.class);

    @Override
    public String supportShape() {
        return "request";
    }

    @PostConstruct
    void init() {
        //允许设置被限制的请求头
        System.setProperty("sun.net.http.allowRestrictedHeaders", "true");
    }

    @Override
    public void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables) {
        CookieContext cookieContext = context.getCookieContext();
        sleep(node, context, variables);
        //重试次数
        int retryCount = NumberUtils.toInt(node.getStringJsonValue(RETRY_COUNT), 0) + 1;
        //重试间隔时间，单位毫秒
        int retryInterval = NumberUtils.toInt(node.getStringJsonValue(RETRY_INTERVAL), 0);
        boolean successed = false;
        for (int i = 0; i < retryCount && !successed; i++) {
            HttpRequest request = HttpRequest.create();
            //设置请求url
            String url = setUrl(node, variables);
            Optional<BloomFilter<String>> bloomFilter = Optional.empty();
            if (Objects.equals("1", node.getStringJsonValue(REPEAT_ENABLE, "0"))) {
                bloomFilter = Optional.ofNullable(createBloomFilter(context));
                if (isDuplicateUrl(bloomFilter, url)) {
                    LOGGER.info("过滤重复URL:{}", url);
                    return;
                }
            }
            SpiderNode root = context.getRootNode();
            List<SpiderNode> nodes = Arrays.asList(root, node);

            //初始化请求
            initRequest(node, context, request, url);
            //设置请求头
            setRequestHeader(nodes, request, context, variables);
            //设置cookie
            setRequestCookies(node, context, variables, cookieContext, request);
            List<InputStream> streams = new ArrayList<>();
            //设置请求参数
            setRequestParams(node, context, variables, request, nodes, streams);
            //设置代理
            setRequestProxy(node, context, variables, request);
            Throwable exception = null;
            try {
                successed = executeRequest(node, variables, cookieContext, request, url, bloomFilter);
            } catch (IOException e) {
                exception = e;
            } finally {
                closeStreams(streams);
                if (!successed) {
                    setRetryAndWriteLog(context, retryCount, retryInterval, i, url, exception);
                }
            }
        }
    }

    private boolean executeRequest(SpiderNode node, Map<String, Object> variables,
                                   CookieContext cookieContext, HttpRequest request,
                                   String url, Optional<BloomFilter<String>> bloomFilter) throws IOException {
        HttpResponse response = request.execute();
        boolean success = response.getStatusCode() == 200;
        if (success) {
            bloomFilter.ifPresent(filter -> filter.put(MD5FunctionExecutor.string(url)));
            String charset = node.getStringJsonValue(RESPONSE_CHARSET);
            if (StringUtils.isNotBlank(charset)) {
                response.setCharset(charset);
                LOGGER.debug("设置response charset:{}", charset);
            }
            //cookie存入cookieContext
            cookieContext.putAll(response.getCookies());
            //结果存入变量
            variables.put("resp", response);
        }
        return success;
    }

    private void setRequestParams(SpiderNode node, SpiderContext context, Map<String, Object> variables, HttpRequest request, List<SpiderNode> nodes, List<InputStream> streams) {
        String bodyType = node.getStringJsonValue(BODY_TYPE);
        if ("raw".equals(bodyType)) {
            String contentType = node.getStringJsonValue(BODY_CONTENT_TYPE);
            request.contentType(contentType);
            try {
                Object requestBody = ExpressionUtils.execute(node.getStringJsonValue(REQUEST_BODY), variables);
                context.pause(node.getNodeId(), PauseEvent.REQUEST_BODY, REQUEST_BODY, requestBody);
                request.data(requestBody);
                LOGGER.info("设置请求Body:{}", requestBody);
            } catch (Exception e) {
                LOGGER.debug("设置请求Body出错", e);
            }
        } else if ("form-data".equals(bodyType)) {
            streams.addAll(setRequestFormParameter(node, request, context, variables));
        } else {
            //设置请求参数
            setRequestParameter(nodes, request, context, variables);
        }
    }

    private void setRequestCookies(SpiderNode node, SpiderContext context,
                                   Map<String, Object> variables, CookieContext cookieContext,
                                   HttpRequest request) {
        SpiderNode root = context.getRootNode();
        //设置全局Cookie
        Map<String, String> rootCookies = getRequestCookie(root, root.getListJsonValue(COOKIE_NAME, COOKIE_VALUE), context, variables);
        if (!rootCookies.isEmpty()) {
            LOGGER.info("设置全局Cookie：{}", rootCookies);
            request.cookies(rootCookies);
        }
        //设置自动管理的Cookie
        boolean cookieAutoSet = !"0".equals(node.getStringJsonValue(COOKIE_AUTO_SET));
        if (cookieAutoSet && !cookieContext.isEmpty()) {
            context.pause(node.getNodeId(), PauseEvent.COOKIE_AUTO_SET, COOKIE_AUTO_SET, cookieContext);
            request.cookies(cookieContext);
            LOGGER.info("自动设置Cookie：{}", cookieContext);
        }
        //设置本节点Cookie
        Map<String, String> nodeCookies = getRequestCookie(node, node.getListJsonValue(COOKIE_NAME, COOKIE_VALUE), context, variables);
        if (!nodeCookies.isEmpty()) {
            request.cookies(nodeCookies);
            LOGGER.debug("设置节点Cookies：{}", nodeCookies);
        }
        if (cookieAutoSet) {
            cookieContext.putAll(nodeCookies);
        }
    }

    private void closeStreams(List<InputStream> streams) {
        for (InputStream is : streams) {
            try {
                is.close();
            } catch (RuntimeException | IOException e) {
                //do nothing
            }
        }
    }

    private void setRetryAndWriteLog(SpiderContext context, int retryCount, int retryInterval,
                                     int i, String url, Throwable exception) {
        if (i + 1 < retryCount) {
            if (retryInterval > 0) {
                DgbObjects.sleep(retryInterval);
            }
            LOGGER.info("第{}次重试:{}", i + 1, url);
        } else {
            writeErrorLogFile(context, url);
            LOGGER.error("请求{}出错,异常信息:{}", url, exception);
        }
    }

    private boolean isDuplicateUrl(Optional<BloomFilter<String>> bloomFilter, String url) {
        return bloomFilter.map(filter -> filter.mightContain(MD5FunctionExecutor.string(url)))
                          .orElse(false);
    }

    private void setRequestProxy(SpiderNode node, SpiderContext context,
                                 Map<String, Object> variables, HttpRequest request) {
        String proxy = node.getStringJsonValue(PROXY);
        if (StringUtils.isNotBlank(proxy)) {
            try {
                Object value = ExpressionUtils.execute(proxy, variables);
                context.pause(node.getNodeId(), PauseEvent.COMMON, PROXY, value);
                if (value != null) {
                    String[] proxyArr = value.toString().split(":");
                    if (proxyArr.length == 2) {
                        request.proxy(proxyArr[0], Integer.parseInt(proxyArr[1]));
                        LOGGER.info("设置代理：{}", proxy);
                    }
                }
            } catch (RuntimeException e) {
                LOGGER.error("设置代理出错，异常信息:{}", e.getMessage(), e);
            }
        }
    }

    private void writeErrorLogFile(SpiderContext context, String url) {
        //记录访问失败的日志
        //测试环境
        if (context.getFlowId() != null) {
            File file = new File(workspace, context.getFlowId() + File.separator + "logs" + File.separator + "access_error.log");
            try {
                File directory = file.getParentFile();
                if (directory.exists() || directory.mkdirs()) {
                    FileUtils.write(file, url + "\r\n", "UTF-8", true);
                }
            } catch (IOException ignored) {
                //do nothing
            }
        }
    }

    private void initRequest(SpiderNode node, SpiderContext context, HttpRequest request, String url) {
        context.pause(node.getNodeId(), PauseEvent.COMMON, URL, url);
        LOGGER.info("设置请求url:{}", url);
        request.url(url);
        //设置请求超时时间
        int timeout = NumberUtils.toInt(node.getStringJsonValue(TIMEOUT), 60000);
        LOGGER.debug("设置请求超时时间:{}", timeout);
        request.timeout(timeout);

        String method = Objects.toString(node.getStringJsonValue(REQUEST_METHOD), "GET");
        //设置请求方法
        request.method(method);
        LOGGER.debug("设置请求方法:{}", method);

        //是否跟随重定向
        boolean followRedirects = !"0".equals(node.getStringJsonValue(FOLLOW_REDIRECT));
        request.followRedirect(followRedirects);
        LOGGER.debug("设置跟随重定向：{}", followRedirects);

        //是否验证TLS证书,默认是验证
        if ("0".equals(node.getStringJsonValue(TLS_VALIDATE))) {
            request.validateTLSCertificates(false);
            LOGGER.debug("设置TLS证书验证：{}", false);
        }
    }

    private String setUrl(SpiderNode node, Map<String, Object> variables) {
        String url;
        try {
            url = ExpressionUtils.execute(node.getStringJsonValue(URL), variables).toString();
        } catch (RuntimeException e) {
            LOGGER.error("设置请求url出错");
            throw e;
        }
        return url;
    }

    private static void sleep(SpiderNode node, SpiderContext context, Map<String, Object> variables) {
        String sleepCondition = node.getStringJsonValue(SLEEP);
        if (StringUtils.isBlank(sleepCondition)) {
            return;
        }
        Object value = ExpressionUtils.execute(sleepCondition, variables);
        if (value == null) {
            return;
        }
        long sleepTime = NumberUtils.toLong(value.toString(), 0L);
        String intern = node.getNodeId().intern();
        try {
            StringLock.tryLock(intern);
            //实际等待时间 = 上次执行时间 + 睡眠时间 - 当前时间
            Long lastExecuteTime = context.get(LAST_EXECUTE_TIME + node.getNodeId(), 0L);
            if (lastExecuteTime != 0) {
                sleepTime = lastExecuteTime + sleepTime - System.currentTimeMillis();
            }
            if (sleepTime > 0) {
                context.pause(node.getNodeId(), PauseEvent.COMMON, SLEEP, sleepTime);
                LOGGER.debug("设置延迟时间:{}ms", sleepTime);
                DgbObjects.sleep(sleepTime);
            }
            context.put(LAST_EXECUTE_TIME + node.getNodeId(), System.currentTimeMillis());
        } catch (InterruptedException t) {
            LOGGER.error("获取锁失败,intern->{}", intern, t);
            Thread.currentThread().interrupt();
        } finally {
            StringLock.unlock(intern);
        }
    }

    public static final String SET_PARAM_LOG = "设置请求参数：{}={}";

    private List<InputStream> setRequestFormParameter(SpiderNode node, HttpRequest request, SpiderContext context,
                                                      Map<String, Object> variables) {
        List<Map<String, String>> parameters = node.getListJsonValue(PARAMETER_FORM_NAME, PARAMETER_FORM_VALUE,
                PARAMETER_FORM_TYPE, PARAMETER_FORM_FILENAME);
        if (parameters == null) {
            return Collections.emptyList();
        }
        return parameters.stream()
                         .filter(nameValue -> StringUtils.isNotBlank(nameValue.get(PARAMETER_FORM_NAME)))
                         .map(nameValue -> {
                             String parameterName = nameValue.get(PARAMETER_FORM_NAME);
                             String parameterValue = nameValue.get(PARAMETER_FORM_VALUE);
                             String parameterType = nameValue.get(PARAMETER_FORM_TYPE);
                             String parameterFilename = nameValue.get(PARAMETER_FORM_FILENAME);
                             try {
                                 boolean hasFile = "file".equals(parameterType);
                                 Object value = ExpressionUtils.execute(parameterValue, variables);
                                 if (hasFile) {
                                     return setRequestFormParameter(node, request, context,
                                             value, parameterName, parameterFilename);
                                 } else {
                                     request.data(parameterName, value);
                                     context.pause(node.getNodeId(), PauseEvent.REQUEST_BODY, parameterName, value);
                                     LOGGER.info(SET_PARAM_LOG, parameterName, value);
                                 }
                             } catch (Exception e) {
                                 LOGGER.error("设置请求参数：{}出错,异常信息:{}", parameterName, e);
                             }
                             return null;
                         }).filter(Objects::nonNull).collect(Collectors.toList());
    }

    private InputStream setRequestFormParameter(SpiderNode node, HttpRequest request, SpiderContext context, Object value,
                                                String parameterName, String parameterFilename) {
        InputStream stream;
        if (value instanceof byte[]) {
            stream = new ByteArrayInputStream((byte[]) value);
        } else if (value instanceof String) {
            stream = new ByteArrayInputStream(((String) value).getBytes());
        } else if (value instanceof InputStream) {
            stream = (InputStream) value;
        } else {
            stream = null;
        }
        if (stream != null) {
            request.data(parameterName, parameterFilename, stream);
            context.pause(node.getNodeId(), PauseEvent.REQUEST_BODY, parameterName, parameterFilename);
            LOGGER.info(SET_PARAM_LOG, parameterName, parameterFilename);
        } else {
            LOGGER.warn("设置请求参数：{}失败，无二进制内容", parameterName);
        }
        return stream;
    }

    private Map<String, String> getRequestCookie(SpiderNode node, List<Map<String, String>> cookies, SpiderContext context, Map<String, Object> variables) {
        Map<String, String> cookieMap = new HashMap<>(8);
        if (cookies == null) {
            return cookieMap;
        }
        for (Map<String, String> nameValue : cookies) {
            Object value;
            String cookieName = nameValue.get(COOKIE_NAME);
            if (StringUtils.isNotBlank(cookieName)) {
                String cookieValue = nameValue.get(COOKIE_VALUE);
                try {
                    value = ExpressionUtils.execute(cookieValue, variables);
                    if (value != null) {
                        cookieMap.put(cookieName, value.toString());
                        context.pause(node.getNodeId(), PauseEvent.REQUEST_COOKIE, cookieName, value.toString());
                        LOGGER.info("设置请求Cookie：{}={}", cookieName, value);
                    }
                } catch (Exception e) {
                    LOGGER.error("设置请求Cookie：{}出错,异常信息：{}", cookieName, e);
                }
            }
        }
        return cookieMap;
    }

    private void setRequestParameter(List<SpiderNode> nodes, HttpRequest request, SpiderContext context,
                                     Map<String, Object> variables) {
        nodes.stream().filter(node -> node.getListJsonValue(PARAMETER_NAME, PARAMETER_VALUE) != null)
             .forEach(node -> node.getListJsonValue(PARAMETER_NAME, PARAMETER_VALUE)
                                  .stream().filter(nameValue -> StringUtils.isNotBlank(nameValue.get(PARAMETER_VALUE)))
                                  .forEach(nameValue -> {
                                      String parameterName = nameValue.get(PARAMETER_NAME);
                                      String parameterValue = nameValue.get(PARAMETER_VALUE);
                                      try {
                                          Object value = ExpressionUtils.execute(parameterValue, variables);
                                          context.pause(node.getNodeId(), PauseEvent.REQUEST_PARAMETER, parameterName, value);
                                          LOGGER.info(SET_PARAM_LOG, parameterName, value);
                                          request.data(parameterName, value);
                                      } catch (Exception e) {
                                          LOGGER.error("设置请求参数：{}出错,异常信息：{}", parameterName, e);
                                          request.data(parameterName, null);
                                      }
                                  }));
    }

    private void setRequestHeader(List<SpiderNode> nodes, HttpRequest request, SpiderContext context,
                                  Map<String, Object> variables) {
        nodes.stream().filter(node -> node.getListJsonValue(HEADER_NAME, HEADER_VALUE) != null)
             .forEach(node -> node.getListJsonValue(HEADER_NAME, HEADER_VALUE)
                                  .stream().filter(nameValue -> StringUtils.isNotBlank(nameValue.get(HEADER_NAME)))
                                  .forEach(nameValue -> {
                                      String headerName = nameValue.get(HEADER_NAME);
                                      String headerValue = nameValue.get(HEADER_VALUE);
                                      try {
                                          Object value = ExpressionUtils.execute(headerValue, variables);
                                          context.pause(node.getNodeId(), PauseEvent.REQUEST_HEADER, headerName, value);
                                          LOGGER.info("设置请求Header：{}={}", headerName, value);
                                          request.header(headerName, value);
                                      } catch (RuntimeException e) {
                                          LOGGER.error("设置请求Header：{}出错,异常信息：{}", headerName, e);
                                          request.header(headerName, null);
                                      }
                                  }));
    }

    @Override
    public List<Grammar> grammars() {
        List<Grammar> grammars = Grammar.findGrammars(SpiderResponse.class, "resp", "SpiderResponse", false);
        Grammar grammar = new Grammar();
        grammar.setFunction("resp");
        grammar.setComment("抓取结果");
        grammar.setOwner("SpiderResponse");
        grammars.add(grammar);
        return grammars;
    }

    @Override
    public void beforeStart(SpiderContext context) {
        //do nothing
    }

    private BloomFilter<String> createBloomFilter(SpiderContext context) {
        BloomFilter<String> filter = context.get(BLOOM_FILTER_KEY);
        if (filter == null) {
            Funnel<CharSequence> funnel = Funnels.stringFunnel(StandardCharsets.UTF_8);
            String fileName = context.getFlowId() + File.separator + "url.bf";
            File file = new File(workspace, fileName);
            if (file.exists()) {
                try (FileInputStream fis = new FileInputStream(file)) {
                    filter = BloomFilter.readFrom(fis, funnel);
                } catch (IOException e) {
                    LOGGER.error("读取布隆过滤器出错", e);
                }
            } else {
                filter = BloomFilter.create(funnel, capacity, errorRate);
            }
            context.put(BLOOM_FILTER_KEY, filter);
        }
        return filter;
    }

    @Override
    public void afterEnd(SpiderContext context) {
        BloomFilter<String> filter = context.get(BLOOM_FILTER_KEY);
        if (filter != null) {
            File file = new File(workspace, context.getFlowId() + File.separator + "url.bf");
            if (!file.getParentFile().exists()) {
                file.getParentFile().mkdirs();
            }
            try (FileOutputStream fos = new FileOutputStream(file)) {
                filter.writeTo(fos);
                fos.flush();
            } catch (IOException e) {
                LOGGER.error("保存布隆过滤器出错", e);
            }
        }
    }
}
